unit QPeerThread;

interface
uses
  Classes, Windows, SysUtils, SyncObjs, IdTCPServer,
  Packet, PacketData, siAuto;

Type
  TQPeerThread = class(TIdPeerThread)
  private
    FPacketRead, FPacketWrite: TPacketData;
    FCSIO, FCSBuffer: TCriticalSection;
    FList: TList;
    FThread: TThread;
    FStream, FStreamWrite: TMemoryStream;

    procedure EventThreadTerminate(Sender:TObject);
    procedure Do_StartThread;
    procedure Do_StopThread;
  protected

    procedure Clear;
    procedure BeforeExecute; override;
    procedure AfterExecute; override;

  public
    procedure EnterCS;
    procedure LeaveCS;
  
    procedure Add(Data:pointer; Size:cardinal);
    function Get(var Data:pointer; var Size:cardinal):Boolean;

    procedure WriteString(cmd:integer; src:String);

    property PacketRead: TPacketData read FPacketRead;
    property PacketWrite: TPacketData read FPacketWrite;
    property Stream: TMemoryStream read FStream;
  published
  end;

  TQueueProcessThread = Class(TThread)
    private
      FPeerThread: TQPeerThread;
      FStream: TMemoryStream;
      procedure Do_Process;
    protected
      procedure Execute; override;
    public
      WaitHandle: THandle;
      Constructor Create(AThread: TQPeerThread);
      Destructor  Destroy; override;
  End;


implementation


{ TQPeerThread }

procedure TQPeerThread.Add(Data: pointer; Size: cardinal);
var
  Packet : TPacket;
begin
  FCSBuffer.Enter;
  try
    Packet := TPacket.Create(Data, Size);
    FList.Add(Packet);
  finally
    FCSBuffer.Leave;
  end;

end;

procedure TQPeerThread.AfterExecute;
begin
  inherited;
  Do_StopThread;
  
  self.Clear;
  FreeAndNil( FList );
  FCSBuffer.Free;
  FCSIO.Free;
  FreeAndNil( FPacketRead );
  FreeAndNil( FPacketWrite );
  FreeAndNil( FStream );
  FreeAndNil( FStreamWrite );
end;

procedure TQPeerThread.BeforeExecute;
begin
  inherited;
  Flist := TList.Create;
  FCSBuffer := TCriticalSection.Create;
  FCSIO     := TCriticalSection.Create;
  FPacketRead   := TPacketData.Create;
  FPacketWrite  := TPacketData.Create;
  FStream := TMemoryStream.Create;
  FStreamWrite := TMemoryStream.Create;

  Do_StartThread;
end;

procedure TQPeerThread.Clear;
var
  Loop : Integer;
  Packet : TPacket;
begin
  FCSBuffer.Enter;
  try
    for Loop := 0 to FList.Count - 1 do begin
      Packet := Pointer(FList[Loop]);
      Packet.Free;
    end;

    FList.Clear;
  finally
    FCSBuffer.Leave;
  end;
end;

procedure TQPeerThread.Do_StartThread;
begin
  if Assigned(Fthread) then Exit;

  FThread := TQueueProcessThread.Create(self);
  FThread.OnTerminate := EventThreadTerminate;
  FThread.Resume;

end;

procedure TQPeerThread.Do_StopThread;
begin
  if not Assigned(FThread) then Exit;
  FThread.OnTerminate := nil;
  FThread.Terminate;
  WaitForSingleObject( TQueueProcessThread(FThread).WaitHandle, 500);
end;

procedure TQPeerThread.EnterCS;
begin
  FCSIO.Enter;
end;

procedure TQPeerThread.EventThreadTerminate(Sender: TObject);
begin
  FThread := nil;
end;

function TQPeerThread.Get(var Data: pointer; var Size: cardinal): Boolean;
var
  Packet : TPacket;
begin
  FCSBuffer.Enter;
  try
    Data := nil;
    Size := 0;
    Result := false;

    if FList.Count = 0 then Exit;

    Result := true;
    Packet := Pointer(FList[0]);
    FList.Delete(0);
    try
      Size := Packet.Size;
      if Size > 0 then begin
        GetMem(Data, Size);
        Move(Packet.Data^, Data^, Size);
      end;
    finally
      Packet.Free;
    end;
  finally
    FCSBuffer.Leave;
  end;
end;

procedure TQPeerThread.LeaveCS;
begin
  FCSIO.Leave;
end;

procedure TQPeerThread.WriteString(cmd: integer; src: String);
begin
  FStream.Clear;
  FStream.Write(src[1], Length(src));
  FStream.Position := 0;
  //siMain.LogStream('Send 1', FStream );

  PacketWrite.Add(cmd, FStream.Memory, FStream.Size );

  self.Connection.WriteLn('command');
//  siMain.LogStream('Send 2', PacketWrite.Stream );

  Self.Connection.WriteStream( PacketWrite.Stream, True, True );

end;


{ TQueueProcessThread }

constructor TQueueProcessThread.Create(AThread: TQPeerThread);
begin
  inherited Create(True);

  FStream := TMemoryStream.Create;
  FPeerThread := AThread;
  self.FreeOnTerminate := True;
  WaitHandle := CreateEvent(nil, False, False, '');

end;

destructor TQueueProcessThread.Destroy;
begin
  CloseHandle( WaitHandle );
  FStream.Free;
  
  inherited;
end;

procedure TQueueProcessThread.Do_Process;
var
  Data:Pointer;
  Size: Cardinal;
begin
  if not FPeerThread.Get(Data, Size) then Exit;

  FPeerThread.EnterCS;
  try
    try
      FStream.Clear;
      FStream.Size := Size;
      Move(Data^, FStream.Memory^, Size);

      FPeerThread.EnterCS;
      try
      FPeerThread.Connection.WriteLn('command');
      FPeerThread.Connection.WriteStream(FStream, True, True);
      finally
        FPeerThread.LeaveCS;
      end;

    finally
      FreeMem( Data );
    end;
  finally
    FPeerThread.LeaveCS;
  end;
end;

procedure TQueueProcessThread.Execute;
begin
  try
    while not Terminated do
    begin
      Do_Process;
      
      WaitforSingleObject(Self.Handle, 1);
    end;
  finally
    SetEvent(WaitHandle);
  end;

end;

end.
